package com.atguigu.chapter08;

import com.atguigu.Bean.UserBehavior;
import lombok.val;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.awt.*;
import java.time.Duration;
import java.util.Date;

/**
 * @ClassName: Flink01_PV_Project
 * @Description:
 * @Author: kele
 * @Date: 2021/4/9 18:52
 *
 *
 *
 *需求：实现一个网站总浏览量的统计。我们可以设置滚动时间窗口，实时统计每小时内的网站PV。
 *
 * 1、获取数据
 * 2、设置水印（采用事件时间）
 * 3、设置滚动窗口
 *
 **/
public class Flink01_PV_Project {

    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",20000);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        env.setParallelism(2);

        env.readTextFile("in/UserBehavior.csv")
                .flatMap(new FlatMapFunction<String, UserBehavior>() {

                    @Override
                    public void flatMap(String line, Collector<UserBehavior> out) throws Exception {

                        String[] split = line.split(",");
                        UserBehavior behavior = new UserBehavior(Long.valueOf(split[0]),
                                                                Long.valueOf(split[1]), Integer.valueOf(split[2]),
                                                                split[3], Long.valueOf(split[4]));

                        if("pv".equals(behavior.getBehavior())){
                            out.collect(behavior);
                        }

                    }
                })
                //添加水印
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner(new SerializableTimestampAssigner<UserBehavior>() {
                            @Override
                            public long extractTimestamp(UserBehavior element, long recordTimestamp) {
                                return element.getTimestamp() * 1000L;
                            }
                        })
                )
                .keyBy(UserBehavior::getBehavior)
                .window(TumblingEventTimeWindows.of(Time.minutes(60)))
                .process(new ProcessWindowFunction<UserBehavior, String, String, TimeWindow>() {

                    private ValueState<Long> pVcount;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        pVcount = getRuntimeContext().getState(new ValueStateDescriptor<Long>("PVcount", Long.class));
                    }

                    @Override
                    public void process(String key,
                                        Context ctx,
                                        Iterable<UserBehavior> elements,
                                        Collector<String> out) throws Exception {


                        pVcount.update(0l);

                        for (UserBehavior element : elements) {

                            this.pVcount.update(this.pVcount.value() + 1);

                        }

                        Date start = new Date(ctx.window().getStart());
                        Date end = new Date(ctx.window().getEnd());

                        out.collect("窗口：["+start+ "---" +end + "),PV数目为： "
                                + this.pVcount.value()
                        );

                    }
                })
                .print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }




    }

}
